package com.xzx.flink.streamapi.watermark;

import com.xzx.flink.bean.ClickEvent;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Random;

/**
 * 断点式的水位线
 *
 * @version 1.0
 * @auther xinzhixuan
 * @date 2022/4/30 13:58
 */
public class Watermark_Source {
    private static final Logger log = LoggerFactory.getLogger(Watermark_Source.class);

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env.addSource(new ClickSourceWithWatermark()).print();

        env.execute(Watermark_Source.class.getSimpleName());
    }

    public static class ClickSourceWithWatermark  implements ParallelSourceFunction<ClickEvent> {

        private boolean running = true;

        @Override
        public void run(SourceContext<ClickEvent> ctx) throws Exception {
            Random random = new Random();
            String[] users = {"Mary", "Alice", "Bob", "Cary"};
            String[] urls = {"./home", "./cart", "./fav", "./prod?id=1","./prod?id=2"};
            while (running) {
                ClickEvent clickEvent = new ClickEvent(users[random.nextInt(users.length)], urls[random.nextInt(urls.length)], System.currentTimeMillis());
                // 跟之前的collect()方法不同的是，这里收集数据的时候已经是带了时间戳了
                ctx.collectWithTimestamp(clickEvent, clickEvent.getTimestamp());
                // 这里也发出了水位线
                ctx.emitWatermark(new org.apache.flink.streaming.api.watermark.Watermark(clickEvent.getTimestamp() - 1));
                Thread.sleep(1000);
            }
        }

        @Override
        public void cancel() {
            running = false;
        }
    }
}
